package com.margo.project.consumer.kafka.service;


import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

import com.margo.project.consumer.common.model.EasydlDiscernMessageVO;



/**
 * Caused by: org.springframework.messaging.converter.MessageConversionException: 
 * Cannot convert from [java.lang.String] to [com.margo.project.consumer.common.model.EasydlDiscernMessageVO] 
 * 
 * https://blog.csdn.net/Crystalqy/article/details/94830862
 * 自定义的消息对象不在kafka信任的包路径下
 * org.apache.kafka.common.errors.SerializationException
 * 
 * Caused by: java.lang.ClassNotFoundException: com.easydl.project.amqp.model.EasydlDiscernMessageVO
 * @author fxb208
 *
 */
@Service
public class KafkaConsumerService {

	Logger logger = LoggerFactory.getLogger(this.getClass());

	 @KafkaListener(topics="easydlDiscern", groupId="easydl-consumer-group")
	 public void messageConsumer(@Payload EasydlDiscernMessageVO
			 easydlDiscernMessageVO,@Headers Map<String,Object> headers) {
		 for(String str:headers.keySet()) {
			 logger.info("kafka:"+str);
		 }
	 }

}
